package com.practice.concurrent;

import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 自定义线程池拒绝策略
 */
public class MyThreadRejectedExecutionHandler implements RejectedExecutionHandler {

    public MyThreadRejectedExecutionHandler(){
        super();
    }

//    /**
//     * Netty中的处理方式
//     * CallerRunsPolicy的思想
//     * 直接创建一个线程池以外的线程处理这些任务，为了保证任务的实时处理，
//     * 这种做法可能需要良好的硬件设备且临时创建的线程无法做到准确的监控
//     * @param r the runnable task requested to be executed
//     * @param executor the executor attempting to execute this task
//     */
//    @Override
//    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
//        try {
//            final Thread t  = new Thread(r,"Temporary task executor");
//            t.start();
//        }catch (Throwable e){
//            throw new RejectedExecutionException("Failed to start a new thread",e);
//        }
//    }


    /**
     * ActiveMQ中的处理方式
     * CallerRunsPolicy的思想
     * 尝试在指定的时效内尽可能的争取将任务入队，以保证最大交付：
     * @param r the runnable task requested to be executed
     * @param executor the executor attempting to execute this task
     */
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        try {
            //限时阻塞等待，实现尽可能的交付
            executor.getQueue().offer(r,60, TimeUnit.SECONDS);
        }catch (InterruptedException e){
            throw new RejectedExecutionException("Interrupted waiting for BrokerService.worker");
        }
        throw new RejectedExecutionException("Timed Out while attempting to enqueue Task");
    }
}
